package pers.mihao.quickstream.common.plus.jdbc.sequence;


import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.sql.*;

/**
 * @author mihao
 * @since 2023/1/3
 */
@Slf4j
public class MysqlSequence extends AbstractSequence {

    private String tableName = "sequence";
    private String sequenceValKeyName = "sequence_val";
    private String gmtModifiedColumnName = "gmt_modified";
    private String nameColumnName = "group_key";

    @Resource
    DataSource dataSource;

    @Resource
    TransactionTemplate transactionTemplate;

    public MysqlSequence(String group) {
        super.sequenceKey = group;
    }

    public MysqlSequence(String group, String tableName, String sequenceValKeyName, String gmtModifiedColumnName,
                         String nameColumnName) {
        super.sequenceKey = group;
        this.tableName = tableName;
        this.sequenceValKeyName = sequenceValKeyName;
        this.gmtModifiedColumnName = gmtModifiedColumnName;
        this.nameColumnName = nameColumnName;
    }

    @PostConstruct
    public void init() {
        String selectSql = getSelectSql();

        try (Connection connection = getConnection(); PreparedStatement preparedStatement = connection.prepareStatement(selectSql)) {
            preparedStatement.setString(1, super.sequenceKey);
            ResultSet resultSet = preparedStatement.executeQuery();
            if (!resultSet.next()) {
                addNewSequence(connection);
            }
            setNewRangeValue();
        } catch (SQLException e) {
            log.error("", e);
            throw new RuntimeException(e);
        }

    }

    private void addNewSequence(Connection connection) {
        transactionTemplate.execute(c -> {
            int value = 0;
            try (PreparedStatement preparedStatement = connection.prepareStatement(getInsertSql())) {
                preparedStatement.setString(1, super.sequenceKey);
                preparedStatement.setInt(2, 0);
                preparedStatement.setDate(3, new Date(System.currentTimeMillis()));
                value = preparedStatement.executeUpdate();
            } catch (SQLException e) {
                log.error("", e);
                throw new RuntimeException(e);
            }
            if (value != 1) {
                throw new SequenceException();
            }
            return c;
        });
    }

    private Connection getConnection() {
        try {
            return dataSource.getConnection();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    protected int getAndIncrementNewRange(String sequenceKey, int innerStep) {
        String sql = getUpdateSql();
        int currValue;
        try (Connection connection = getConnection(); PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
            currValue = updateNewSeqVal(connection, preparedStatement);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

        return currValue;
    }

    private int updateNewSeqVal(Connection connection, PreparedStatement preparedStatement) throws SQLException {
        for (; ; ) {
            int currValue = getCurrValue(connection);
            preparedStatement.setInt(1, currValue + innerStep);
            preparedStatement.setDate(2, new Date(System.currentTimeMillis()));
            preparedStatement.setString(3, super.sequenceKey);
            preparedStatement.setInt(4, currValue);
            int updateSuccess = preparedStatement.executeUpdate();
            if (updateSuccess == 1) {
                return currValue;
            }
        }
    }

    private int getCurrValue(Connection connection) {
        try (PreparedStatement preparedStatement = connection.prepareStatement(getSelectSql())) {
            preparedStatement.setString(1, super.sequenceKey);
            ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                return resultSet.getInt(getValueColumnName());
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        throw new SequenceException();
    }


    protected String getInsertSql() {
        return "insert into " + getTableName() + "(" +
                getNameColumnName() + "," +
                getValueColumnName() + "," +
                getGmtModifiedColumnName() + ") values(?,?,?);";
    }

    protected String getSelectSql() {
        StringBuilder buffer = new StringBuilder();
        buffer.append("select ").append(getValueColumnName());
        buffer.append(" from ").append(getTableName());
        buffer.append(" where ").append(getNameColumnName()).append(" = ?");
        return buffer.toString();
    }

    protected String getUpdateSql() {
        StringBuilder buffer = new StringBuilder();
        buffer.append("update ").append(getTableName());
        buffer.append(" set ").append(getValueColumnName()).append(" = ?, ");
        buffer.append(getGmtModifiedColumnName()).append(" = ? where ");
        buffer.append(getNameColumnName()).append(" = ? and ");
        buffer.append(getValueColumnName()).append(" = ?");
        return buffer.toString();
    }

    private String getTableName() {
        return tableName;
    }

    private String getNameColumnName() {
        return nameColumnName;
    }

    private String getValueColumnName() {
        return sequenceValKeyName;
    }

    private String getGmtModifiedColumnName() {
        return gmtModifiedColumnName;
    }
}
